package com.qiandw.common.cassandra

import java.lang.reflect.Field
import java.text.SimpleDateFormat
import java.util.Map.Entry
import java.util._
import java.util.function.Consumer

import com.datastax.driver.core._
import com.datastax.driver.core.querybuilder.{QueryBuilder, Update}
import com.datastax.driver.core.utils.UUIDs
import com.datastax.driver.mapping.annotations.Column
import com.datastax.driver.mapping.{Mapper, MappingManager, Result}
import org.springframework.beans.factory.annotation.{Autowired, Value}

import scala.collection.JavaConversions._

/**
  * cassandra 范型DAO 打包通用方法
  *
  * @author Liang Dong
  */
abstract class AbstractCassandraDao {

  @Value("${db.test.cassandra.defaultKeyspace}")
  var keyspace: String = _

  private val fieldMapperMap: java.util.Map[String, FieldMapper] = new java.util.HashMap[String, FieldMapper]

  private val DATE_KEY_FORMATS = new java.util.ArrayList[String]
  DATE_KEY_FORMATS.add("yyyy")
  DATE_KEY_FORMATS.add("yyyyMM")
  DATE_KEY_FORMATS.add("yyyyMMdd")
  DATE_KEY_FORMATS.add("yyyyMMddHH")

  private val MAX_RETRY_COUNT = 50

  private lazy val mappingManager: MappingManager = new MappingManager(Session)

  @Autowired
  protected var sf: KeyspaceSessionFactory = _

  private val psMap = new java.util.HashMap[String, PreparedStatement]

  protected def Session: Session = sf.getKeyspaceSession(keyspace)

  protected def Prepare(key: String, query: String) {
    psMap.put(key, Session.prepare(query))
  }

  protected def Execute(key: String, values: Object*): ResultSet = {
    Session.execute(psMap.get(key).bind(values: _*))
  }

  def Execute(statement: Statement): ResultSet = {
    Session.execute(statement)
  }

  /**
    * 执行一条语句，如果遇到异常则重试
    */
  def ExecuteWithExceptionRetry(statement: Statement): ResultSet = {
    var wasNotSuccessful = true
    val rs: ResultSet = null
    while (wasNotSuccessful) {
      try {
        Session.execute(statement).wasApplied()
        wasNotSuccessful = false
      } catch {
        case t: Throwable =>
      }
    }
    rs
  }

  def getTableNames: java.util.List[String] = {
    val list = new java.util.LinkedList[String]
    Session.getCluster.getMetadata.getKeyspace(keyspace).getTables.forEach(new Consumer[TableMetadata] {
      override def accept(t: TableMetadata): Unit = {
        list.add(t.getName)
      }
    })
    list
  }

  def getViewNames: java.util.List[String] = {
    val list = new java.util.LinkedList[String]
    Session.getCluster.getMetadata.getKeyspace(keyspace).getMaterializedViews.foreach(v => {
      list.add(v.getName)
    })
    list
  }

  protected def updateQueryMathVersion(table: String, matchingVersion: Long, versionFieldName: String): Update = {
    val update = QueryBuilder.update(table)
    update.onlyIf(QueryBuilder.eq(versionFieldName, matchingVersion))
    update.`with`(QueryBuilder.set(versionFieldName, matchingVersion + 1))
    update
  }

  protected def versionUpdateQuery(table: String, matchingVersion: Long, versionFieldName: String): UpdateQueryWrapper = {
    val update = QueryBuilder.update(table)
    update.onlyIf(QueryBuilder.eq(versionFieldName, matchingVersion))
    update.`with`(QueryBuilder.set(versionFieldName, matchingVersion + 1))
    new UpdateQueryWrapper(update)
  }

  implicit def updateQueryWrapperToUpdate(wrapper: UpdateQueryWrapper): Update = wrapper.update

  class UpdateQueryWrapper(val update: Update) {

    def where(field: String, value: Object): UpdateQueryWrapper = {
      this.update.where(QueryBuilder.eq(field, value))
      this
    }

    def set(field: String, value: Object): UpdateQueryWrapper = {
      this.update.`with`(QueryBuilder.set(field, value))
      this
    }
  }

  /**
    * 为prepared statment绑定参数并且返回绑定后的statment
    */
  protected def BindStatment(key: String, values: Object*): Statement = psMap.get(key).bind(values: _*)

  protected def RandomUUID: UUID = UUID.randomUUID

  protected def TimeUUID: UUID = UUIDs.timeBased()

  protected def DateKeys(date: Date): java.util.Set[String] = {
    val result = new java.util.HashSet[String]
    for (i <- 0 until DATE_KEY_FORMATS.size) {
      result.add(new SimpleDateFormat(DATE_KEY_FORMATS.get(i)).format(date))
    }
    result
  }


  protected def untilSuccess(payload: (() => Boolean)): Boolean = {
    var retryCount = 0
    var wasApplied = false
    while (!wasApplied && retryCount < MAX_RETRY_COUNT) {
      try {
        wasApplied = payload()
      } catch {
        case t: Throwable =>
      }
      if (!wasApplied) {
        retryCount += 1
      }
    }
    wasApplied
  }

  private def getFieldMapper[T](klass: Class[T]): FieldMapper = {
    val key = classKey(klass)
    var fm = fieldMapperMap.get(key)
    if (fm == null) {
      mapFields(klass)
    }
    fm = fieldMapperMap.get(key)
    fm
  }

  private def mapFields[T](klass: Class[T]): Unit = {
    fieldMapperMap.synchronized({
      val key = classKey(klass)
      if (!fieldMapperMap.containsKey(key)) {
        val fieldMapper = new FieldMapper
        klass.getDeclaredFields.foreach(f => {
          val columDef = f.getAnnotation(classOf[Column])
          if (columDef != null) {
            f.setAccessible(true)
            fieldMapper.fieldMap.put(columDef.name(), f)
          }
        })
        fieldMapperMap.put(key, fieldMapper)
      }
    })
  }

  private def classKey[T](klass: Class[T]): String = klass.getName

  /*protected*/ def getMapper[T](t: Class[T]): Mapper[T] = {
    mappingManager.mapper(t)
  }

  /**
    * 直接将语句映射为对象，取出所有元素
    *
    * @param statment 语句
    * @param t        对象类
    * @tparam T 对象模板
    */
  def map[T](statment: Statement, t: Class[T]): java.util.List[T] = getMapper(t).map(Execute(statment)).all()

  /**
    * 获取一个对象
    *
    * @param t   对象类型
    * @param ids ID序列, 按顺序排列
    * @return 对象
    */
  def get[T](t: Class[T], ids: Object*): T = {
    getMapper(t).get(ids: _*)
  }

  /**
    * 保存一个对象
    */
  def save[T](obj: T, t: Class[T]): Unit = {
    Session.execute(getMapper(t).saveQuery(obj)).one()
  }

  /**
    * 获取保存的语句
    */
  def saveQuery[T](obj: T, t: Class[T]): Statement = {
    getMapper(t).saveQuery(obj)
  }

  def saveIfNotExists[T](obj: T, t: Class[T]): Boolean = {
    val mapper = getMapper(t)
    val fieldMapper = getFieldMapper(t)
    val cql = QueryBuilder.insertInto(mapper.getTableMetadata).ifNotExists()
    fieldMapper.fieldMap.entrySet().forEach(new Consumer[Entry[String, Field]] {
      override def accept(entry: Entry[String, Field]): Unit = {
        val value = entry.getValue.get(obj)
        if (value != null) {
          cql.value(entry.getKey, value)
        }
      }
    })
    var wasSuccessful = false
    untilSuccess(() => {
      wasSuccessful = Execute(cql).wasApplied()
      true
    })
    wasSuccessful
  }

  /**
    * 删除对象
    */
  def del[T](t: Class[T], ids: Object*): Unit = {
    getMapper(t).delete(ids: _*)
  }

  /**
    * 删除对象
    */
  def del(obj: Object): Unit = {
    getMapper(obj.getClass).delete(obj)
  }

  /**
    * 将结果队列转换为对象列表
    */
  def map[T](results: ResultSet, t: Class[T]): Result[T] = {
    getMapper(t).map(results)
  }

  /**
    * 获取默认的keyspace
    */
  def getDefaultKeyspace: String = keyspace
}

